/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package miner

import (
	"context"
	"encoding/hex"
	"fmt"
	"net/http"
	"time"

	"hundsun.com/hsl/hschain/common/types"

	"github.com/golang/protobuf/proto"
	"go.uber.org/atomic"
	"hundsun.com/hsl/hschain/common/keystore"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/util/pb"
	dposcomm "hundsun.com/hsl/hschain/consensus/impl/dpos/common"
	minercomm "hundsun.com/hsl/hschain/consensus/impl/dpos/miner/common"
	consensustypes "hundsun.com/hsl/hschain/consensus/types"
	"hundsun.com/hsl/hschain/protos"
	"hundsun.com/hsl/hschain/protos/config"
	"hundsun.com/hsl/hschain/protos/consensus/dpos"
	pbledger "hundsun.com/hsl/hschain/protos/ledger"
	"hundsun.com/hsl/hschain/protos/seqcenter"
)

// Producer 负责出块
type Producer struct {
	module *DposMiner
	ctx    context.Context

	// endorsementPeers 背书节点对应的PeerID列表
	endorsementPeers map[protos.Address]string
	// sequenceCenterPeer 序列化中心PeerID
	sequenceCenterPeer string
	// edBlocksBuffer miner在本轮出的块及相应的背书
	edBlocksBuffer *EdBlocksBuffer
	// producedBlockEdFailed miner在本轮出的块是否有背书失败的
	producedBlockEdFailed  *atomic.Bool
	currentRound           *atomic.Int64
	currentRoundMinerIndex *atomic.Int32

	currentRoundCreatedBlocks int
}

func (p *Producer) init(minerAccount keystore.Account, producingInterval, pendingTxPoolRetryInterval, checkBlockExecResultRetryInterval int) error {
	var err error
	producerAdapterConf := &consensustypes.ProducerConfig{
		ModuleLog:                            p.module.Log,
		MinerAccount:                         minerAccount,
		ProducingInterval:                    producingInterval,
		MinPendingTxFetchInterval:            pendingTxPoolRetryInterval,
		MinCheckBlockExecResultFetchInterval: checkBlockExecResultRetryInterval,
	}
	p.endorsementPeers = make(map[protos.Address]string, dposcomm.MaxMinersCount)
	p.currentRound = atomic.NewInt64(-1)
	p.currentRoundMinerIndex = atomic.NewInt32(-1)
	p.edBlocksBuffer = NewEdBlockBuffer(p.module.GetClient().GetTopicClient(), p.module.Log)

	//判断在此时间片是否该继续出块，出现如下情况则停止出块：
	//1) 当前节点miner存在于出块miner列表中
	//2) 本时间片的块背书失败了;
	//3) 区块尚未同步。当前轮次中，轮次开始/出块/背书时也会检测（与序列号中心比对），若发现已同步，则本轮不再与序列号中中心比对。
	//4) 本轮已经出了６块
	//5) 时间片过了
	shouldProceed := func() bool {
		conType := p.module.baseClient.GetCurParam().GetConsensus().GetType()
		switch conType {
		case config.DposConsensusType:
			bExisted := false
			currentMiners := p.module.baseClient.GetCurParam().GetConsensus().GetDposPayload().GetDefaultMiners()
			for _, miner := range currentMiners {
				if miner.Address == string(p.module.MinerAccount.GetAddress()) {
					bExisted = true
				}
			}
			if !bExisted {
				p.module.Log.Info("abort producing due to current miner %s not existed in miner list", p.module.MinerAccount.GetAddress())
				return false
			}
		default:
			p.module.Log.Warnf("abort producing due to consensus type already update to %s", conType)
			return false
		}

		blockEdFailed := p.producedBlockEdFailed.Load()
		if blockEdFailed {
			p.module.Log.Warn("abort producing due to blockEdFailure")
			return false
		}

		blockSynchronized := p.module.IsBlockSynchronized()
		if !blockSynchronized {
			p.module.Log.Warn("abort producing due to block not synchronized")
			return false
		}

		if p.currentRoundCreatedBlocks >= dposcomm.MaxBlocksInTimeSlice {
			p.module.Log.Debugf("abort producing due to %d blocks created", p.currentRoundCreatedBlocks)
			return false
		}

		_, producePeriodEnd := dposcomm.CalculateProducePeriod(p.currentRound.Load(), int(p.currentRoundMinerIndex.Load()))
		timeSliceElapsed := time.Now().UnixNano()/int64(time.Millisecond) >= producePeriodEnd
		if timeSliceElapsed {
			p.module.Log.Debugf("abort producing due to time slice elapsed")
			return false
		}

		return true
	}

	beforeBlockCreatedFunc := func() (lastBlockNumber uint64, lastBlockHash protos.Hash) {
		lastBlock := p.module.CurrentRoundBlockBuffer.GetLast()
		if lastBlock == nil {
			lastBlockCheckedMsg := p.module.GetClient().GetLastBlockCheckedMessage()
			lastBlockNumber = lastBlockCheckedMsg.BlockNumber
			lastBlockHash = lastBlockCheckedMsg.BlockHash
			p.module.Log.Tracef("try to produce with prev block[number='%d',lastBlockHash=%s] in BlockCheckedMessage", lastBlockNumber, hex.EncodeToString(lastBlockHash))
		} else {
			lastBlockNumber = lastBlock.Header.Number
			lastBlockHash = lastBlock.Hash()
			p.module.Log.Tracef("try to produce with  prev block[number='%d',lastBlockHash=%s] in RoundBlocksBuffer", lastBlockNumber, hex.EncodeToString(lastBlockHash))
		}
		return
	}

	afterBlockCreatedFunc := func(newBlock *pbledger.Block) {
		newBlockHash := newBlock.Hash()
		blkLog := p.module.Log.WithFields(map[string]interface{}{
			logging.LogFieldBlockHash:   newBlockHash.GetShortString(),
			logging.LogFieldBlockNumber: newBlock.Header.Number,
		})
		p.module.CurrentRoundBlockBuffer.Add(newBlock, p.module.Log)
		_, currentMiners := minercomm.GetCurrentRoundInfo()
		p.edBlocksBuffer.Put(NewEdBlock(newBlock, currentMiners, p.module.Log))
		p.module.GetClient().PutPreRemovedTxHashesIntoLocalCache(newBlockHash, pb.GetTxHashesInBlock(newBlock), blkLog)
		blkLog.Info("request endorsements from other miners")
		for _, peerID := range p.endorsementPeers {
			blkLog.Tracef("Send endorsement request to peer[%s]", peerID)
			err := p.module.sendEndorsementReq(peerID, newBlock)
			if err != nil {
				blkLog.Error("send endorsement request failed, err: %s\n", err)
			}
		}
		p.currentRoundCreatedBlocks++
	}

	err = p.module.baseClient.InitProducer(producerAdapterConf, shouldProceed, beforeBlockCreatedFunc, afterBlockCreatedFunc)
	if err != nil {
		return err
	}

	return nil
}

//Run 出块
func (p *Producer) Run() {
	p.module.Log.Infof("started dpos producer loop...")
	defer p.module.Log.Info("exit dpos producer loop...")
	minerAddress := p.module.MinerAccount.GetAddress()

	for {
		var minerIndex int
		var roundStartUnixMilli int64
		var currentRound int64
		start := time.Now()
		p.module.Log.WithField(dposcomm.LogFieldRound, p.currentRound.Load()).Debug("start to sync miners of next round...")
		for {
			var syncedRound int64
			var err error
			nowUnixMillis := time.Now().UnixNano() / int64(time.Millisecond)
			currentRound, roundStartUnixMilli, err = dposcomm.CalculateRound(nowUnixMillis)
			if err != nil {
				panic(err)
			}

			syncedRound, minerIndex = minercomm.GetCurrentRoundMinerIndex(minerAddress)
			if currentRound == syncedRound {
				p.currentRound.Store(currentRound)
				p.currentRoundMinerIndex.Store(int32(minerIndex))
				break
			}
		}
		rLog := p.module.Log.WithField(dposcomm.LogFieldRound, currentRound)
		rLog.WithCostField(start).Debugf("round synced.round start time:%d", roundStartUnixMilli)

		if minerIndex >= 0 {
			producePeriodStart, _ := p.prepareForProducing(rLog)
			sleepMillis := producePeriodStart - time.Now().UnixNano()/int64(time.Millisecond)
			if sleepMillis > 0 {
				rLog.Debugf("will sleep %dms before producing", sleepMillis)
				select {
				case <-time.After(time.Duration(sleepMillis) * time.Millisecond):
					p.produceBlocks(currentRound)
				case <-p.ctx.Done():
					p.module.ShutdownWg.Done()
					return
				}
			}
		}

		nextRoundStartMillis := roundStartUnixMilli + dposcomm.EpochIntervalMillis
		sleepMillis := nextRoundStartMillis - time.Now().UnixNano()/int64(time.Millisecond)
		rLog.Debugf("will sleep %dms util next round start", sleepMillis)
		select {
		case <-time.After(time.Duration(sleepMillis) * time.Millisecond):
			continue
		case <-p.ctx.Done():
			return
		}

	}
}

// updateEdClients 更新当前miner与其他miner建立的长链接
func (p *Producer) updateEdClients() {
	_, currentMiners := minercomm.GetCurrentRoundInfo()
	for addr, v := range currentMiners {
		p.endorsementPeers[addr] = v.PeerID
	}
}

// prepareForProducing 做本轮出块前的准备工作，返回出块时间段[producePeriodStart, producePeriodEnd)
func (p *Producer) prepareForProducing(rLog logging.Log) (producePeriodStart, producePeriodEnd int64) {
	//1. 清理工作
	p.module.CurrentRoundBlockBuffer.Clear(p.module.Log)
	p.edBlocksBuffer.Clear()
	p.producedBlockEdFailed.Store(false)
	p.module.ClearBlockSynchronized()
	//触发是否同步的检查
	_ = p.module.IsBlockSynchronized()

	//2. 获取可以标记交易的锁（背书时）
	markerLock := fmt.Sprintf(txMarkerLockFmt, p.currentRound.Load())
	rLog.Debugf("try to acquire '%s' lock", markerLock)
	p.module.TxMarkerLockHoldFlag.Store(true)
	rLog.Infof("got '%s' lock:%v", markerLock, true)

	//3. 建立与其他miner的连接，用于背书
	producePeriodStart, producePeriodEnd = dposcomm.CalculateProducePeriod(p.currentRound.Load(), int(p.currentRoundMinerIndex.Load()))
	rLog.Debugf("produce time [%d,%d)", producePeriodStart, producePeriodEnd)
	nowUnixMillis := time.Now().UnixNano() / int64(time.Millisecond)
	if nowUnixMillis >= producePeriodEnd {
		rLog.Info("time slice passed by and no need to connect to other miners")
		return
	}

	p.updateEdClients()
	return
}

func (p *Producer) produceBlocks(currentRound int64) {
	rLog := p.module.Log.WithField(dposcomm.LogFieldRound, currentRound)
	rLog.Debugf("produceBlocks entry...")

	// 0. 清空当前miner已生产块缓存
	p.edBlocksBuffer.Clear()
	p.currentRoundCreatedBlocks = 0
	p.module.baseClient.ProduceBlocks()
}

func (p *Producer) recvEndorse(originPeerID string, msg *consensustypes.ConsensRelayMsg) error {
	var resp = &dpos.BlockEndorsementResponse{}
	proto.Unmarshal(msg.Payload, resp)
	bLog := p.module.Log.WithField(dposcomm.LogFieldRound, resp.BlockNumber)
	blkEd := p.edBlocksBuffer.Get(resp.BlockNumber)
	if blkEd == nil {
		bLog.Warn("not found in edBlocksBuffer")
		return types.ErrNotFound
	}
	bLog.Infof("received ed response from peer: %s", originPeerID)

	blkEdStatus, firstAchieved := blkEd.Put(resp)
	if blkEdStatus == EdBlockStatusFailed {
		p.producedBlockEdFailed.Store(true)
		if firstAchieved {
			bLog.WithField(logging.LogFieldCost, time.Now().UnixNano()/int64(time.Millisecond)-blkEd.block.Header.BlockTime).Warn("failed to get enough eds")
		}
	} else if blkEdStatus == EdBlockStatusPassed {
		if !firstAchieved {
			bLog.Trace("ed response was ignored,because block ed result had been determined")
			return nil
		}
		bLog.WithField(logging.LogFieldCost, time.Now().UnixNano()/int64(time.Millisecond)-blkEd.block.Header.BlockTime).Infof("get enough eds")

		// TODO
		// 如果本地blockSequenceServer不为空，直接放入blockQueue队列中
		// 如果本地blockSequenceServer为空，通过p2p进行广播
		if p.module.GetClient().GetSequenceServer() != nil {
			p.edBlocksBuffer.TryToSendPassedBlocksToSeqCenter(p.sequenceCenterPeer, p.module.GetClient().GetSequenceServer().GetBlockQueue(), true)
		} else {
			p.edBlocksBuffer.TryToSendPassedBlocksToSeqCenter(p.sequenceCenterPeer, nil, false)
		}
	}
	return nil
}

// SeqBlockSavedCallback 区块通过序列化中心签名后的回调
func (p *Producer) SeqBlockSavedCallback(originPeerID string, msg *consensustypes.ConsensRelayMsg) error {
	var resp = &seqcenter.SaveBlockResponse{}
	proto.Unmarshal(msg.Payload, resp)

	bLog := p.module.Log.WithField(logging.LogFieldBlockNumber, resp.BlockNumber)
	bLog.Infof("Receive sequence center save block resp from peerID: %s", originPeerID)
	blkEd := p.edBlocksBuffer.Get(resp.BlockNumber)
	if blkEd == nil {
		bLog.Warn("block not found int edBlocksBuffer")
		return types.ErrNotFound
	}

	if resp.StatusCode != http.StatusOK {
		bLog.Errorf("failed to get block sequence:%s", resp.ErrorMessage)
		return types.ErrNotFound
	}
	if !dposcomm.VerifyBlockSequenceSignature(blkEd.block, resp.BlockSequence, resp.Signature, p.module.Log) {
		bLog.Error("failed to validate block sequence signature")
		return types.ErrInvalidSignature
	}
	blkEd.block.Header.Number = resp.BlockNumber
	blkEd.block.Header.BlockSequence = resp.BlockSequence
	blkEd.block.Header.BlockSequenceSignature = resp.Signature

	err := p.module.GetClient().WriteBlock(blkEd.block)
	if err != nil {
		bLog.Errorf("dispatched new block to blockchain module failed, err:%s\n.", err.Error())
		return err
	}
	bLog.Debug("dispatched new block to blockchain module.")
	return p.module.GetClient().BroadCastBlock(blkEd.block, 5)
}
